KAFKA-19662: Reset share group offsets for unsubscribed topics by AndrewJSchofield · Pull Request #20708 · apache/kafka

AndrewJSchofield · 2025-10-15T19:31:42Z

This PR allows the kafka-share-groups.sh --reset-offsets tool to be used
to set offsets for topics which are not currently subscribed in a share
group. It also works if the share group does not yet exist. This brings
the capability in line with the equivalent function in
Kafka-consumer-groups.sh. The primary purpose is to allow offsets to be
set before the share group is first used as a way of initialising in a
known state.

Reviewers: Jimmy Wang 48462172+JimmyWang6@users.noreply.github.com,
Kuan-Po Tseng brandboat@gmail.com, Apoorv Mittal
apoorvmittal10@gmail.com

JimmyWang6

LGTM, thanks for the change! Just two small things left.

JimmyWang6 · 2025-10-16T07:42:58Z

    ) {
-        List<CoordinatorRecord> records = new ArrayList<>();
-        ShareGroup group = groupMetadataManager.shareGroup(groupId);
-        group.validateOffsetsAlterable();


I think the method validateOffsetsAlterable() could be removed.

JimmyWang6 · 2025-10-16T07:47:08Z

+            } catch (ExecutionException ee) {
+                Throwable cause = ee.getCause();
+                if (cause instanceof GroupIdNotFoundException) {
+                    resetOffsetsForInactiveGroup(groupId);


Just wondering, why is there a retry mechanism when GroupIdNotFoundException is thrown? Is this for better reliability? I see ConsumerGroupCommand also does this.

I think this PR aims to align the behavior between consumer groups and shared consumer groups.

For regular consumer groups, we already have similar logic — if the group doesn’t exist, it will be automatically created. see

kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java

Lines 465 to 466 in fd9b551

log.info("[GroupId {}] Creating a simple consumer group via manual offset commit.", request.groupId());

group = groupMetadataManager.getOrMaybeCreateClassicGroup(request.groupId(), true);

In this PR, the shared consumer group now follows the same behavior: https://github.com/apache/kafka/pull/20708/files#diff-00f0f81cf13e66781777d94f7d2e68a581663385c37e98792507f2294c91bb09R8309-R8310

Back to your question, I think your understanding is correct. This is something like a retry mechanism: if describeShareGroups throws a GroupIdNotFoundException, then will call resetOffsetsForInactiveGroup again, which will implicitly create the shared group.

Perhaps Andrew can elaborate more 😃

It's because if you do the operation on a non-existent group, it creates the group (which is by definition empty) and then resets the offsets. I know this is a bit weird, but if you want to initialise the group in a particular state before starting any consumers, this is how you would do it.

brandboat

It seems the failed test testAlterShareGroupOffsets is related to this patch. I also wasn’t able to get it to pass in my local environment.

By the way, do you think we might need to add a test for the tools change in this PR as well?

brandboat · 2025-10-16T13:08:03Z

+            return partitionsToReset;
+        }
+
+        private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId, Collection<TopicPartition> partitionsToReset) {


nit: now groupId become unused, could you please remove this variable? thanks!

AndrewJSchofield · 2025-10-16T17:53:30Z

It seems the failed test testAlterShareGroupOffsets is related to this patch. I also wasn’t able to get it to pass in my local environment.

By the way, do you think we might need to add a test for the tools change in this PR as well?

Yes, I agree that the test failure is related. I'm going to fix that test and probably add another few. The fact that so little failed is a good sign that overall this way of addressing the problem works well.

brandboat

LGTM, thanks!

JimmyWang6

Thanks! LGTM.

apoorvmittal10

Thanks for the PR, LGTM!

…e#20708) This PR allows the kafka-share-groups.sh --reset-offsets tool to be used to set offsets for topics which are not currently subscribed in a share group. It also works if the share group does not yet exist. This brings the capability in line with the equivalent function in Kafka-consumer-groups.sh. The primary purpose is to allow offsets to be set before the share group is first used as a way of initialising in a known state. Reviewers: Jimmy Wang <48462172+JimmyWang6@users.noreply.github.com>, Kuan-Po Tseng <brandboat@gmail.com>, Apoorv Mittal <apoorvmittal10@gmail.com>

* Revert "Merge pull request #161 from showuon/cl3_ut" This reverts commit 55efc8d, reversing changes made to 1c8b6d3. * Revert "Merge pull request #155 from showuon/cl3_share_auto_create" This reverts commit 05f8874, reversing changes made to 143e6cc. * MINOR: Cleanups in Tools Module (3/n) (apache#20332) This PR aims at cleaning up the tools module further by getting rid of some extra code which can be replaced by `record` Reviewers: Chia-Ping Tsai <chia7712@gmail.com> * MINOR: kafka-stream-groups.sh should fail quickly if the partition leader is unavailable (apache#20271) This PR applies the same partition leader check for `StreamsGroupCommand` as `ShareGroupCommand` and `ConsumerGroupCommand` to avoid the command execution timeout. Reviewers: Lucas Brutschy <lucasbru@apache.org> * KAFKA-19662: Allow resetting offset for unsubscribed topic in kafka-share-groups.sh (apache#20453) The `kafka-share-groups.sh` tool checks whether a topic already has a start-offset in the share group when resetting offsets. This is not necessary. By removing the check, it is possible to set a start offset for a topic which has not yet but will be subscribed in the future, thus initialising the consumption point. There is still a small piece of outstanding work to do with resetting the offset for a non-existent group which should also create the group. A subsequent PR will be used to address that. Reviewers: Jimmy Wang <48462172+JimmyWang6@users.noreply.github.com>, Lan Ding <isDing_L@163.com>, Apoorv Mittal <apoorvmittal10@gmail.com> * KAFKA-19662: Reset share group offsets for unsubscribed topics (apache#20708) This PR allows the kafka-share-groups.sh --reset-offsets tool to be used to set offsets for topics which are not currently subscribed in a share group. It also works if the share group does not yet exist. This brings the capability in line with the equivalent function in Kafka-consumer-groups.sh. The primary purpose is to allow offsets to be set before the share group is first used as a way of initialising in a known state. Reviewers: Jimmy Wang <48462172+JimmyWang6@users.noreply.github.com>, Kuan-Po Tseng <brandboat@gmail.com>, Apoorv Mittal <apoorvmittal10@gmail.com> --------- Co-authored-by: Sanskar Jhajharia <sjhajharia@confluent.io> Co-authored-by: jimmy <wangzhiwang611@gmail.com> Co-authored-by: Andrew Schofield <aschofield@confluent.io>

KAFKA-19662: Reset share group offsets for unsubscribed topics

85d1c20

github-actions Bot added tools group-coordinator labels Oct 15, 2025

AndrewJSchofield added KIP-932 Queues for Kafka ci-approved labels Oct 15, 2025

JimmyWang6 reviewed Oct 16, 2025

View reviewed changes

brandboat reviewed Oct 16, 2025

View reviewed changes

Tests and review comments

6e39ce5

github-actions Bot added the core Kafka Broker label Oct 17, 2025

brandboat approved these changes Oct 18, 2025

View reviewed changes

AndrewJSchofield requested a review from apoorvmittal10 October 19, 2025 10:58

JimmyWang6 approved these changes Oct 21, 2025

View reviewed changes

apoorvmittal10 approved these changes Oct 23, 2025

View reviewed changes

AndrewJSchofield merged commit 6d16f68 into apache:trunk Oct 23, 2025
25 checks passed

AndrewJSchofield deleted the KAFKA-19662-1 branch October 23, 2025 17:37

	log.info("[GroupId {}] Creating a simple consumer group via manual offset commit.", request.groupId());
	group = groupMetadataManager.getOrMaybeCreateClassicGroup(request.groupId(), true);

Uh oh!

Conversation

AndrewJSchofield commented Oct 15, 2025 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

JimmyWang6 left a comment

Choose a reason for hiding this comment

Uh oh!

JimmyWang6 Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield Oct 17, 2025

Choose a reason for hiding this comment

Uh oh!

JimmyWang6 Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

brandboat Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

brandboat left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

brandboat Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield commented Oct 16, 2025

Uh oh!

brandboat left a comment

Choose a reason for hiding this comment

Uh oh!

JimmyWang6 left a comment

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

AndrewJSchofield commented Oct 15, 2025 •

edited by github-actions Bot

Loading

brandboat left a comment •

edited

Loading